package com.datagateway.component;

import com.datagateway.model.ProcessedData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Redis数据源实现
 * 
 * @author Data Gateway Team
 * @version 1.0.0
 */
public class RedisDataSource implements DataSourceManager.DataSourceInstance {

    private static final Logger logger = LoggerFactory.getLogger(RedisDataSource.class);

    private final Map<String, Object> config;
    private final String type = "REDIS";
    private JedisPool jedisPool;
    private boolean connected = false;

    public RedisDataSource(Map<String, Object> config) {
        this.config = new ConcurrentHashMap<>(config);
        initialize();
    }

    /**
     * 初始化Redis连接
     */
    private void initialize() {
        try {
            String host = (String) config.getOrDefault("host", "localhost");
            int port = (Integer) config.getOrDefault("port", 6379);
            String password = (String) config.get("password");
            int database = (Integer) config.getOrDefault("database", 0);
            int timeout = (Integer) config.getOrDefault("timeout", 2000);

            // 配置连接池
            JedisPoolConfig poolConfig = new JedisPoolConfig();
            poolConfig.setMaxTotal(100);
            poolConfig.setMaxIdle(20);
            poolConfig.setMinIdle(5);
            poolConfig.setTestOnBorrow(true);
            poolConfig.setTestOnReturn(true);
            poolConfig.setTestWhileIdle(true);

            // 创建连接池
            if (password != null && !password.isEmpty()) {
                jedisPool = new JedisPool(poolConfig, host, port, timeout, password, database);
            } else {
                jedisPool = new JedisPool(poolConfig, host, port, timeout, null, database);
            }

            connected = true;
            logger.info("Redis数据源初始化成功: {}:{}", host, port);

        } catch (Exception e) {
            logger.error("Redis数据源初始化失败: {}", e.getMessage(), e);
            connected = false;
        }
    }

    @Override
    public boolean testConnection() {
        try (Jedis jedis = jedisPool.getResource()) {
            String result = jedis.ping();
            connected = "PONG".equals(result);
            return connected;
        } catch (Exception e) {
            logger.error("Redis连接测试失败: {}", e.getMessage(), e);
            connected = false;
            return false;
        }
    }

    @Override
    public boolean writeData(List<ProcessedData> dataList) {
        if (!connected || jedisPool == null) {
            logger.error("Redis连接未建立");
            return false;
        }

        if (dataList == null || dataList.isEmpty()) {
            return true;
        }

        try (Jedis jedis = jedisPool.getResource()) {
            String keyPrefix = (String) config.getOrDefault("keyPrefix", "datagateway:");
            String dataType = (String) config.getOrDefault("dataType", "hash");

            for (ProcessedData data : dataList) {
                String key = keyPrefix + data.getId();
                
                if ("hash".equals(dataType)) {
                    // 使用Hash存储
                    Map<String, String> hash = new HashMap<>();
                    hash.put("id", data.getId());
                    hash.put("content", data.getContent());
                    hash.put("timestamp", String.valueOf(data.getTimestamp()));
                    hash.put("metadata", convertMetadataToString(data.getMetadata()));
                    jedis.hset(key, hash);
                } else if ("string".equals(dataType)) {
                    // 使用String存储JSON
                    String json = convertProcessedDataToJson(data);
                    jedis.set(key, json);
                } else if ("list".equals(dataType)) {
                    // 使用List存储
                    String json = convertProcessedDataToJson(data);
                    jedis.lpush(keyPrefix + "list", json);
                }
                
                // 设置过期时间
                int expireSeconds = (Integer) config.getOrDefault("expireSeconds", 3600);
                if (expireSeconds > 0) {
                    jedis.expire(key, expireSeconds);
                }
            }

            logger.info("Redis数据写入成功: count={}", dataList.size());
            return true;

        } catch (Exception e) {
            logger.error("Redis数据写入失败: {}", e.getMessage(), e);
            return false;
        }
    }

    @Override
    public List<ProcessedData> readData(String query, Map<String, Object> parameters) {
        if (!connected || jedisPool == null) {
            logger.error("Redis连接未建立");
            return Collections.emptyList();
        }

        List<ProcessedData> resultList = new ArrayList<>();

        try (Jedis jedis = jedisPool.getResource()) {
            String keyPrefix = (String) config.getOrDefault("keyPrefix", "datagateway:");
            String dataType = (String) config.getOrDefault("dataType", "hash");

            if (query == null || query.trim().isEmpty()) {
                // 查询所有键
                Set<String> keys = jedis.keys(keyPrefix + "*");
                for (String key : keys) {
                    ProcessedData data = readDataByKey(jedis, key, dataType);
                    if (data != null) {
                        resultList.add(data);
                    }
                }
            } else {
                // 根据查询条件查找
                String key = keyPrefix + query;
                ProcessedData data = readDataByKey(jedis, key, dataType);
                if (data != null) {
                    resultList.add(data);
                }
            }

            logger.info("Redis数据读取成功: count={}", resultList.size());
            return resultList;

        } catch (Exception e) {
            logger.error("Redis数据读取失败: {}", e.getMessage(), e);
            return Collections.emptyList();
        }
    }

    @Override
    public boolean executeSql(String sql, Map<String, Object> parameters) {
        if (!connected || jedisPool == null) {
            logger.error("Redis连接未建立");
            return false;
        }

        try (Jedis jedis = jedisPool.getResource()) {
            // Redis不支持SQL，这里可以执行一些Redis命令
            if (sql.toLowerCase().startsWith("flush")) {
                jedis.flushDB();
                logger.info("Redis数据库清空成功");
                return true;
            } else if (sql.toLowerCase().startsWith("info")) {
                String info = jedis.info();
                logger.info("Redis信息: {}", info);
                return true;
            } else {
                logger.warn("Redis不支持SQL语句: {}", sql);
                return false;
            }

        } catch (Exception e) {
            logger.error("Redis命令执行失败: sql={}, error={}", sql, e.getMessage(), e);
            return false;
        }
    }

    @Override
    public boolean isConnected() {
        try (Jedis jedis = jedisPool.getResource()) {
            String result = jedis.ping();
            return "PONG".equals(result);
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public void close() {
        try {
            if (jedisPool != null) {
                jedisPool.close();
            }
            connected = false;
            logger.info("Redis数据源连接已关闭");
        } catch (Exception e) {
            logger.error("关闭Redis连接失败: {}", e.getMessage(), e);
        }
    }

    @Override
    public String getType() {
        return type;
    }

    @Override
    public Map<String, Object> getConfig() {
        return new HashMap<>(config);
    }

    /**
     * 根据键读取数据
     */
    private ProcessedData readDataByKey(Jedis jedis, String key, String dataType) {
        try {
            if ("hash".equals(dataType)) {
                Map<String, String> hash = jedis.hgetAll(key);
                if (hash.isEmpty()) {
                    return null;
                }
                return convertHashToProcessedData(hash);
            } else if ("string".equals(dataType)) {
                String json = jedis.get(key);
                if (json == null) {
                    return null;
                }
                return convertJsonToProcessedData(json);
            } else if ("list".equals(dataType)) {
                String json = jedis.lpop(key);
                if (json == null) {
                    return null;
                }
                return convertJsonToProcessedData(json);
            }
            return null;
        } catch (Exception e) {
            logger.error("读取Redis数据失败: key={}, error={}", key, e.getMessage(), e);
            return null;
        }
    }

    /**
     * 将ProcessedData转换为JSON字符串
     */
    private String convertProcessedDataToJson(ProcessedData data) {
        try {
            StringBuilder json = new StringBuilder();
            json.append("{");
            json.append("\"id\":\"").append(data.getId()).append("\",");
            json.append("\"content\":\"").append(data.getContent()).append("\",");
            json.append("\"timestamp\":").append(data.getTimestamp()).append(",");
            json.append("\"metadata\":").append(convertMetadataToString(data.getMetadata()));
            json.append("}");
            return json.toString();
        } catch (Exception e) {
            logger.error("转换ProcessedData为JSON失败: {}", e.getMessage(), e);
            return "{}";
        }
    }

    /**
     * 将JSON字符串转换为ProcessedData
     */
    private ProcessedData convertJsonToProcessedData(String json) {
        try {
            ProcessedData data = new ProcessedData();
            // 简单的JSON解析，实际项目中可以使用Jackson等库
            json = json.trim();
            if (json.startsWith("{") && json.endsWith("}")) {
                json = json.substring(1, json.length() - 1);
                String[] pairs = json.split(",");
                for (String pair : pairs) {
                    String[] keyValue = pair.split(":");
                    if (keyValue.length == 2) {
                        String key = keyValue[0].trim().replaceAll("\"", "");
                        String value = keyValue[1].trim().replaceAll("\"", "");
                        
                        switch (key) {
                            case "id":
                                data.setId(value);
                                break;
                            case "content":
                                data.setContent(value);
                                break;
                            case "timestamp":
                                data.setTimestamp(Long.parseLong(value));
                                break;
                            case "metadata":
                                data.setMetadata(parseMetadataFromString(value));
                                break;
                        }
                    }
                }
            }
            return data;
        } catch (Exception e) {
            logger.error("转换JSON为ProcessedData失败: {}", e.getMessage(), e);
            return null;
        }
    }

    /**
     * 将Hash转换为ProcessedData
     */
    private ProcessedData convertHashToProcessedData(Map<String, String> hash) {
        ProcessedData data = new ProcessedData();
        data.setId(hash.get("id"));
        data.setContent(hash.get("content"));
        data.setTimestamp(Long.parseLong(hash.getOrDefault("timestamp", "0")));
        data.setMetadata(parseMetadataFromString(hash.get("metadata")));
        return data;
    }

    /**
     * 将metadata转换为字符串
     */
    private String convertMetadataToString(Map<String, Object> metadata) {
        if (metadata == null || metadata.isEmpty()) {
            return "{}";
        }
        
        try {
            StringBuilder json = new StringBuilder("{");
            boolean first = true;
            for (Map.Entry<String, Object> entry : metadata.entrySet()) {
                if (!first) {
                    json.append(",");
                }
                json.append("\"").append(entry.getKey()).append("\":");
                if (entry.getValue() instanceof String) {
                    json.append("\"").append(entry.getValue()).append("\"");
                } else {
                    json.append(entry.getValue());
                }
                first = false;
            }
            json.append("}");
            return json.toString();
        } catch (Exception e) {
            logger.error("转换metadata为字符串失败: {}", e.getMessage(), e);
            return "{}";
        }
    }

    /**
     * 从字符串解析metadata
     */
    private Map<String, Object> parseMetadataFromString(String metadataStr) {
        Map<String, Object> metadata = new HashMap<>();
        if (metadataStr == null || metadataStr.trim().isEmpty() || "{}".equals(metadataStr)) {
            return metadata;
        }
        
        try {
            metadataStr = metadataStr.trim();
            if (metadataStr.startsWith("{") && metadataStr.endsWith("}")) {
                metadataStr = metadataStr.substring(1, metadataStr.length() - 1);
                String[] pairs = metadataStr.split(",");
                for (String pair : pairs) {
                    String[] keyValue = pair.split(":");
                    if (keyValue.length == 2) {
                        String key = keyValue[0].trim().replaceAll("\"", "");
                        String value = keyValue[1].trim().replaceAll("\"", "");
                        metadata.put(key, value);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("解析metadata字符串失败: {}", e.getMessage(), e);
        }
        
        return metadata;
    }
}
